﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using WorkStation.Frameworks.Core;
using WorkStation.Frameworks.Core.Extensions;

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Content;

namespace WorkStation.Frameworks.EasyBus.MessageQueue.RabbitMq
{
    /// <summary>
    /// RabbitMq 终结点实现
    /// </summary>
    public class RabbitMqChannel : ChannelBase
    {
        protected RabbitMqEndpoint _endpoint = null;
        protected IConnection _conn = null;
        protected IModel _model = null;
        protected QueueingBasicConsumer _consumer = null;

        public RabbitMqChannel(RabbitMqEndpoint endpoint) 
        {
            this._endpoint = endpoint;

            this._conn = this._endpoint.RabbitMqFactory.CreateConnection();
            this._model = this._conn.CreateModel();
            this._consumer = new QueueingBasicConsumer(this._model);
        }

        public override void WriteMsg(Message msg)
        {
            this._model.ExchangeDeclare(RabbitMqEndpoint.EXCHANGE_NAME, "topic");

            this._model.QueueDeclare(this._endpoint.QueueName, true, false, false, null);

            this._model.QueueBind(this._endpoint.QueueName, RabbitMqEndpoint.EXCHANGE_NAME, msg.RouteKey);

            IMessageBuilder builder = this.EasyBusMessageToRabbitMqMessage(msg, this._model);

            IBasicProperties header = builder.GetContentHeader() as IBasicProperties;

            header.DeliveryMode = 2;

            this._model.BasicPublish(RabbitMqEndpoint.EXCHANGE_NAME, msg.RouteKey, header, builder.GetContentBody());
        }

        public override Message ReadMsg()
        {
            try
            {
                this._model.QueueDeclare(this._endpoint.QueueName, true, false, false, null);

                this._model.BasicConsume(this._endpoint.QueueName, true, this._consumer);

                BasicDeliverEventArgs res = new BasicDeliverEventArgs();
            
                // 使用 this._consumer.Queue.DequeueNoWait(null) ; 造成 erl.exe 进程内存暴涨
                if (this._consumer.Queue.Dequeue(3000, out res))
                {
                    string messageId = (res.BasicProperties.Headers["MessageId"] as byte[]).ToStringByUTF8();
                    string routeKey = (res.BasicProperties.Headers["RouteKey"] as byte[]).ToStringByUTF8();
                    string isRecordingHistory = (res.BasicProperties.Headers["IsRecordingHistory"] as byte[]).ToStringByUTF8();
                    string bodyJson = res.Body.ToStringByUTF8();

                    // 使用回应感觉会降低性能，在没出现丢失消息的情况下，先不使用
                    //this._consumer.Model.BasicAck(res.DeliveryTag, false);

                    return new Message(messageId, bodyJson, routeKey) { IsRecordingHistory = Convert.ToBoolean(isRecordingHistory) };
                }
            }
            catch (Exception ce) 
            {
                // RabbitMq 在接受消息时报 NotSupportedException 异常，不影响正常使用
                if (!(ce is NotSupportedException)) 
                {
                    throw;
                }
            }

            return null;
        }

        public override void Dispose()
        {
            if (this._model != null) 
            {
                this._model.Dispose();
            }

            if (this._conn != null) 
            {
                this._conn.Dispose();
            }
        }

        protected virtual IMessageBuilder EasyBusMessageToRabbitMqMessage(Message msg, IModel model) 
        {
            IMapMessageBuilder msgBuilder = new MapMessageBuilder(model);

            msgBuilder.Headers.Add("MessageId", msg.MessageId);
            msgBuilder.Headers.Add("RouteKey", msg.RouteKey);
            msgBuilder.Headers.Add("IsRecordingHistory", msg.IsRecordingHistory.ToString());

            byte[] body = msg.BodyJson.ToBytesByUTF8();

            msgBuilder.BodyStream.Write(body, 0, body.Length);

            return msgBuilder;
        }
    }
}
